package crossjob;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

/**
 * Created by zhaopeng on 2/7/17.
 */
public class CrossJobReducer extends Reducer<Text, IntWritable, Text, Text> {
    private Text writeValue = new Text();
    private Set<Integer> tags = new HashSet<>();
    private int[] educateTag = {11271, 11272, 11273, 11274};
    private Set<Integer> eduSet = new HashSet<>();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        for (int i : educateTag) {
            eduSet.add(i);
        }
    }

    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        tags.clear();
        boolean isEducateShow = false;
        boolean isCross = false;

        for (IntWritable tmp : values) {
            int nums = tmp.get();
            if (0 == nums) {
                isCross = true;
            } else {
                if (eduSet.contains(nums)) {
                    if (!isEducateShow) {
                        tags.add(nums);
                        isEducateShow = true;
                    }
                } else {
                    tags.add(nums);
                }
            }
        }// end of for

        if (isCross && tags.size() > 0) {
            String result = StringUtils.join(tags, ",");
            writeValue.set(result);
            context.write(key, writeValue);
        }

    }
}
